package com.todo.strategy.pipeline.application.job;

import com.alibaba.fastjson.JSONObject;
import com.todo.common.utils.DateUtils;
import com.todo.common.utils.StringUtils;
import com.todo.strategy.pipeline.application.job.entity.FailedRecordExecuteEntity;
import com.todo.strategy.pipeline.common.enums.ExecuteStatusEnum;
import com.todo.strategy.pipeline.common.enums.StrategyActionEnum;
import com.todo.strategy.pipeline.infrastructure.entity.po.PipelineRecordPO;
import com.todo.strategy.pipeline.infrastructure.repository.PipelineRecordRepository;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author azhebuxing
 * @date 2024/5/8 15:06
 * @description
 */
@Slf4j
@Component
public class StrategyFailedRecordJob {

    @Resource
    private PipelineRecordRepository pipelineRecordRepository;


    /**
     * 线程池执行
     */
    ThreadPoolExecutor executor = new ThreadPoolExecutor(30, 30, 10000, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1000));


    @XxlJob("StrategyFailedRecordJob")
    public void execute(String param) {
        log.info("StrategyFailedRecordJob#execute 管道失败记录重试执行 开始执行 param:【{}】", param);

        if (StringUtils.isNotBlank(param)) {
            FailedRecordExecuteEntity executeEntity = JSONObject.parseObject(param, FailedRecordExecuteEntity.class);
            List<PipelineRecordPO> pipelineRecordPOList = pipelineRecordRepository.queryPipelineRecordByTimeRange(
                    StrategyActionEnum.getByType(executeEntity.getActionType())
                    , ExecuteStatusEnum.getByType(executeEntity.getStatus())
                    , DateUtils.dateTime(DateUtils.YYYY_MM_DD_HH_MM_SS, executeEntity.getStartTime())
                    , DateUtils.dateTime(DateUtils.YYYY_MM_DD_HH_MM_SS, executeEntity.getEndTime()));

            if (CollectionUtils.isEmpty(pipelineRecordPOList)) {
                return;
            }
            for (PipelineRecordPO pipelineRecordPO : pipelineRecordPOList) {
                CompletableFuture.runAsync(() -> doExecute(pipelineRecordPO), executor);
            }
        } else {
            LocalDateTime localDateTime = LocalDateTime.now().plusDays(1);
            List<PipelineRecordPO> failedRecordPOList = pipelineRecordRepository.queryPipelineRecordByTimeRange(
                    StrategyActionEnum.PLATFORM_STRATEGY
                    , ExecuteStatusEnum.FAILED
                    , DateUtils.toDate(localDateTime)
                    , new Date());
            if (CollectionUtils.isEmpty(failedRecordPOList)) {
                return;
            }
            for (PipelineRecordPO pipelineRecordPO : failedRecordPOList) {
                CompletableFuture.runAsync(() -> doExecute(pipelineRecordPO), executor);
            }
        }

    }

    /**
     * 具体执行
     *
     * @param pipelineRecordPO
     * @return
     */
    private void doExecute(PipelineRecordPO pipelineRecordPO) {
        return;
    }

}

